package com.etl.dataflow.executor.communication;

import com.etl.dataflow.executor.core.TaskRunner;
import com.etl.dataflow.executor.entity.ExecutorTaskInfo;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author dx
 * @since 2022/4/21
 */
public class TaskRunnerHolder {
    private static final TaskRunnerHolder INSTANCE = new TaskRunnerHolder();
    private static final int QUEUE_MAX_SIZE = 50;

    private final Map<String, TaskRunner> taskRunnerMap = new ConcurrentHashMap<>(10);

    private final Queue<ExecutorTaskInfo> timeoutTaskQueue = new LinkedList<>();

    private TaskRunnerHolder() {
    }

    public static TaskRunnerHolder getInstance() {
        return INSTANCE;
    }

    public void addTaskRunner(String taskId, TaskRunner taskRunner) {
        taskRunnerMap.put(taskId, taskRunner);
    }

    public Optional<TaskRunner> getTaskRunner(String taskId) {
        TaskRunner taskRunner = this.taskRunnerMap.get(taskId);
        return Optional.ofNullable(taskRunner);
    }

    public void removeTaskRunner(String taskId) {
        this.taskRunnerMap.remove(taskId);
    }

    public List<TaskRunner> getAllTaskRunner() {
        Collection<TaskRunner> runners = taskRunnerMap.values();
        List<TaskRunner> taskRunners = new ArrayList<>();
        for (TaskRunner runner : runners) {
            if (runner != null) {
                taskRunners.add(runner);
            }
        }
        return taskRunners;
    }

    public void removeIllegalTasks() {
        for (Map.Entry<String, TaskRunner> entry : taskRunnerMap.entrySet()) {
            String key = entry.getKey();
            TaskRunner taskRunner = entry.getValue();
            boolean isIllegal = taskRunner.isShutdown() && (taskRunner.getWriter() == null || taskRunner.getReader() == null);
            if (isIllegal) {
                ExecutorTaskInfo executorTaskInfo = taskRunner.getExecutorTaskInfo();
                // 删除此任务
                taskRunnerMap.remove(key);
                // 将任务信息放入队列，消费者会获取之，并通知调度中心任务已结束
                if(timeoutTaskQueue.size() < QUEUE_MAX_SIZE) {
                    timeoutTaskQueue.add(executorTaskInfo);
                }
            }
        }
    }

    public Queue<ExecutorTaskInfo> getTimeoutTaskQueue() {
        return this.timeoutTaskQueue;
    }

    public int getRunningTaskNumber() {
        return taskRunnerMap.size();
    }
}
